#ifndef __M_CHANNEL_H__
#define __M_CHANNEL_H__
#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "../MQcommon/logger.hpp"
#include "../MQcommon/helper.hpp"
#include "../MQcommon/mq_msg.pb.h"
#include "../MQcommon/mq_proto.pb.h"
#include "mq_cconsumer.hpp"
#include <iostream>
#include <mutex>
#include <condition_variable>
#include <unordered_map>

namespace bitmq{
    
    typedef std::shared_ptr<google::protobuf::Message> MessagePtr;
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using basicConsumeResponsePtr = std::shared_ptr<basicConsumeResponse>;
    using basicCommonResponsePtr = std::shared_ptr<basicCommonResponse>;
    using ProtobufCodecPtr =std::shared_ptr<ProtobufCodec>;
    class ClientChannel{

        public:
        using ptr=std::shared_ptr<ClientChannel>;
        ClientChannel(const muduo::net::TcpConnectionPtr& conn, const ProtobufCodecPtr& codec):
        _cid(UUIDHelper::uuid()), _conn(conn), _codec(codec) {}
        bool openChannel(){
            bitmq::openChannelRequest req;
            std::string rid=UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            _codec->send(_conn,req);
            basicCommonResponsePtr ret=waitResponse(rid);
            return ret->ok();
        }
        void closeChannel(){
            closeChannelRequest req;
            std::string rid=UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            _codec->send(_conn,req);
            basicCommonResponsePtr ret = waitResponse(rid);
            if(ret->ok()){
                DLOG("关闭信道[%s]!",_cid.c_str());
            }
            return ;
        }
        bool declareExchange(
            const std::string &name,
            ExchangeType type, 
            bool durable, 
            bool auto_delete,
            google::protobuf::Map<std::string, std::string> &args) {
            //构造一个声明虚拟机的请求对象，
            std::string rid = UUIDHelper::uuid();
            declareExchangeRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(name);
            req.set_exchange_type(type);
            req.set_durable(durable);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);
            //然后向服务器发送请求
            _codec->send(_conn, req);
            //等待服务器的响应
            basicCommonResponsePtr resp = waitResponse(rid);
            //返回
            return resp->ok();
        }
        void deleteExchange(const std::string &name) {
            std::string rid = UUIDHelper::uuid();
            deleteExchangeRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(name);
            _codec->send(_conn, req);
            waitResponse(rid);
            return ;
        }

        bool declareQueue(
            const std::string &qname, 
            bool qdurable, 
            bool qexclusive,
            bool qauto_delete,
            google::protobuf::Map<std::string, std::string> &qargs) {
            std::string rid = UUIDHelper::uuid();
            declareQueueRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(qname);
            req.set_durable(qdurable);
            req.set_auto_delete(qauto_delete);
            req.set_exclusive(qexclusive);
            req.mutable_args()->swap(qargs);
            _codec->send(_conn, req);
            basicCommonResponsePtr resp = waitResponse(rid);
            return resp->ok();
        }
        void deleteQueue(const std::string &qname) {
            std::string rid = UUIDHelper::uuid();
            deleteQueueRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(qname);
            _codec->send(_conn, req);
            waitResponse(rid);
            return ;
        }
        //发送消息
        void basicPublish(const std::string& ename,const std::string& body,const BasicProperties*bp){
            std::string rid = UUIDHelper::uuid();
            basicPublishRequest req;
                req.set_rid(rid);
                req.set_cid(_cid);
                req.set_body(body);
                req.set_exchange_name(ename);
                if (bp != nullptr) {
                    req.mutable_properties()->set_id(bp->id());
                    req.mutable_properties()->set_deliver_mode(bp->deliver_mode());
                    req.mutable_properties()->set_routing_key(bp->routing_key());
                }
                _codec->send(_conn, req);
                basicCommonResponsePtr ret = waitResponse(rid);
                if(ret->ok()){
                    DLOG("发送消息[%s]成功",body.c_str());
                }
        }
        bool queueBind(
            const std::string &ename, 
            const std::string &qname, 
            const std::string &key) {
            std::string rid = UUIDHelper::uuid();
            queueBindRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_queue_name(qname);
            req.set_binding_key(key);
            _codec->send(_conn, req);
            basicCommonResponsePtr resp = waitResponse(rid);
            return resp->ok();
        }

        void queueUnBind(const std::string &ename, const std::string &qname) {
            std::string rid = UUIDHelper::uuid();
            queueUnBindRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_queue_name(qname);
            _codec->send(_conn, req);
            waitResponse(rid);
            return ;
        }
        void basicAck(const std::string &msgid) {
            if (_consumer.get() == nullptr) {
                DLOG("消息确认时，找不到消费者信息！");
                return ;
            }
            std::string rid = UUIDHelper::uuid();
            basicAckRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(_consumer->qname);
            req.set_message_id(msgid);
            _codec->send(_conn, req);
            waitResponse(rid);
            return;
        }
        //订阅队列
        bool basicConsume(const std::string& ctag,const std::string& qname,bool ack_flag,ConsumerCallback cb){
             if(_consumer.get() != nullptr){
                DLOG("已经订阅其他队列消息")
                return false;
             }
             basicConsumeRequest req;
             std::string rid=UUIDHelper::uuid();
             req.set_rid(rid);
             req.set_cid(_cid);
             req.set_consumer_tag(ctag);
             req.set_auto_ack(ack_flag);
             req.set_queue_name(qname);
             _codec->send(_conn,req);
             basicCommonResponsePtr ret=waitResponse(rid);
             if(ret->ok() == true){
                DLOG("队列[%s]订阅成功,当前信道[%s]成为新的消费者",qname.c_str(),_cid.c_str());
             }
             _consumer=std::make_shared<Consumer>(ctag,qname,ack_flag,cb); 
             return ret->ok();    
        }
        //取消当前消费者的订阅
        void basicCancel(){
            if(_consumer.get() == nullptr){
                DLOG("当前消费者未订阅队列");
            }
            basicCancelRequest req;
            std::string rid=UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_consumer_tag(_consumer->tag);
            req.set_queue_name(_consumer->qname);
            _codec->send(_conn,req);
            basicCommonResponsePtr ret = waitResponse(rid);
        }
        public:
        void putbasicCommonResponse(const basicCommonResponsePtr& res){
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = res_map.find(res->rid());
            if( it != res_map.end() ){
                printf("请求在\n");
                return ;
            }
            res_map.insert(std::make_pair(res->rid(),res));
            _cv.notify_all();
        }
        //处理推送而来的消息,直接执行消费者的回调函数
        void consume(const basicConsumeResponsePtr& res){
            DLOG("开始处理,信道id:%s",_cid.c_str())
            if(_consumer.get() == nullptr){
                DLOG("该信道不是消费者")
                return ;
            }
            if(_consumer->tag != res->consumer_tag()){
                DLOG("该信道不是指定的消费者")
            }
            _consumer->callback(res->consumer_tag(),res->mutable_properties(),res->body());
        }
        std::string cid(){
            if (!_cid.empty())
            {
                return _cid;
            }
            return " ";
        }
        private:
        basicCommonResponsePtr waitResponse(std::string& rid){
            std::unique_lock<std::mutex> lock(_mutex);
            //条件变量_cv持有锁进行等待
            _cv.wait(lock,[this,rid](){return res_map.find(rid) != res_map.end(); });
            //出来，代表已经收到请求
            basicCommonResponsePtr rse=res_map[rid];
            res_map.erase(rid);
            return rse;
        }
        private:
        std::string _cid;//信道id，用于标识信道的唯一性
        std::mutex _mutex;//用来保证线程安全
        std::condition_variable _cv;
        muduo::net::TcpConnectionPtr _conn;//连接，用于网络通信的指向
        ProtobufCodecPtr _codec;//protobuf解释器，用于发送消息
        Consumer::ptr _consumer;//消费者，因为当前信道和消费者的关系是一对一，所以信道即是消费者！
        std::unordered_map<std::string,basicCommonResponsePtr> res_map;//用来表示那些请求已经收到了响应，没有收到响应的请求所在的线程，应当阻塞
    };
    class ChannelManager {
        public:
            using ptr = std::shared_ptr<ChannelManager>;
            ChannelManager(){}
            ClientChannel::ptr create(const muduo::net::TcpConnectionPtr &conn, 
                const ProtobufCodecPtr &codec) {
                std::unique_lock<std::mutex> lock(_mutex);
                auto channel = std::make_shared<ClientChannel>(conn, codec);
                _channels.insert(std::make_pair(channel->cid(), channel));
                return channel;
            }
            void remove(const std::string &cid) {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _channels.find(cid);
                if(it == _channels.end()){
                    DLOG("该信道[%s]不存在!",cid.c_str());
                    return ;
                }
                // it->second->closeChannel();
                _channels.erase(cid);
            }
            ClientChannel::ptr get(const std::string &cid) {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _channels.find(cid);
                if (it == _channels.end()) {
                    return ClientChannel::ptr();
                }
                return it->second;
            }
        private:
            std::mutex _mutex;
            std::unordered_map<std::string, ClientChannel::ptr> _channels;
    };
}

#endif